/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.boarsoft.message.test.activemq.amqp;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.boarsoft.common.util.RandomUtil;
import com.boarsoft.message.bean.Message;

public class PublisherTest1 {
	private static final Logger log = LoggerFactory.getLogger(PublisherTest1.class);

	public static void main(String[] args) throws Exception {
//		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(//
//				"http://localhost:8080?trace=true");
//		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(//
//				"tcp://localhost:61616");
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(//
				"amqp://localhost:5672");
		// JmsConnectionFactory factory = new JmsConnectionFactory(//
		// "amqp://localhost:5672");
		// JmsConnectionFactory factory = new
		// JmsConnectionFactory("tcp://localhost:61616");
		Connection connection = factory.createConnection("admin", "admin");
		// connection.setClientID("ebc");
		connection.start();
		final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// Destination destination = session.createQueue("sms_debit");
		Destination destination = session.createQueue("queue1");
		final MessageProducer producer = session.createProducer(destination);
		producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		try {
			int tc = 1;
			ExecutorService es = Executors.newFixedThreadPool(tc);
			final CountDownLatch cdl = new CountDownLatch(tc);
			long l = System.currentTimeMillis();
			for (int n = 0; n < tc; n++) {
				es.execute(new Runnable() {
					public void run() {
						testQueue(session, producer, 1);
						cdl.countDown();
					}
				});
			}
			try {
				cdl.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println(System.currentTimeMillis() - l);
		} finally {
			// producer.send(session.createTextMessage("SHUTDOWN"));
			// Thread.sleep(1000 * 5);
			// connection.close();
		}
		// System.exit(0);
	}

	public static void testQueue(Session session, MessageProducer producer, int n) {
		StringBuilder sb = new StringBuilder();
		sb.append("10000000|*|0211  |*|01|*|20180102|*|152342|*|5810|*|0101|*|           "
				+ "|*|6210330210000139092         |*| 000 |*| 000 |*|000000117049481.00|"
				+ "*|000000017049481.00|*|01 |*|2 |*|0 |*|XHCZ-DWXJC6 中文                      2018-01-02 15:23:42.043|"
				+ "*|                          |*|                                                                                                                         |"
				+ "*|58011200465002 |*|0 |*||*|1|*|6210330210000139092        \n<EOF>\n");
		String content = sb.toString();
		try {
			long l = System.currentTimeMillis();
			for (int i = 0; i < n; i++) {
				Message m = new Message();
				// m.setTarget("sms_debit");
				m.setTarget("queue1");
				m.addRecipient("queue1Consumer1");
				m.setContent(content);
				m.setId(RandomUtil.genUUID());
				TextMessage tm = session.createTextMessage(m.toJson());
				producer.send(tm);
				// ObjectMessage om = session.createObjectMessage(m);
				// producer.send(om);
			}
			log.info("Queue {} messages in {}ms", n, System.currentTimeMillis() - l);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}